package gokv

import (
	"fmt"
	"gitee.com/Cauchy_AQ/gokv/data"
	"gitee.com/Cauchy_AQ/gokv/fio"
	"gitee.com/Cauchy_AQ/gokv/index"
	"gitee.com/Cauchy_AQ/gokv/utils"
	"github.com/gofrs/flock"
	"io"
	"os"
	"path/filepath"
	"slices"
	"strconv"
	"strings"
	"sync"
	"time"
)

const (
	// seqNoKey 事务序列号文件记录的 key 值
	seqNoKey = "seqNo.key"

	// fileLockName 数据目录文件锁名称
	fileLockName = "flock"
)

// DB 存储引擎实例
type DB struct {
	opt             *Options                  // 配置项
	mtx             *sync.RWMutex             // 互斥锁
	activeFile      *data.DataFile            // 活跃文件，可写
	olderFiles      map[uint32]*data.DataFile // 旧文件数据，只读
	index           index.Indexer             // 内存索引
	seqNo           uint64                    // 全局事务递增序列号
	isMerging       bool                      // 合并数据文件标识
	fileLock        *flock.Flock              // 文件锁保证多进程之间互斥
	bytesWrite      uint                      // 累积写了多少字节
	ticker          *time.Ticker              // 打点器，定时持久化文件
	invalidDataSize int64                     // 有多少无效的数据
}

// Stat 存储引擎统计信息
type Stat struct {
	KeyNum            uint  // key 总数量
	DataFileNum       uint  // 数据文件数量
	ReclaimableSize   int64 // 可以进行 merge 回收的数据量，单位字节
	AvailableDiskSize int64 // 数据磁盘剩余可用空间大小
	DataDiskSize      int64 // 数据目录所占磁盘空间大小
}

// Open 打开引擎实例
func Open(option *Options) (*DB, error) {
	// 校验配置
	if err := checkOptions(option); err != nil {
		return nil, err
	}

	// 判断当前数据目录是否正在使用，获取文件锁
	fileLock := flock.New(filepath.Join(option.DirPath, fileLockName))
	hold, err := fileLock.TryLock()
	if err != nil {
		return nil, err
	}
	if !hold {
		return nil, ErrDatabaseIsUsing
	}

	// 创建引擎
	db := &DB{
		opt:        option,
		mtx:        new(sync.RWMutex),
		olderFiles: make(map[uint32]*data.DataFile),
		index:      index.NewIndexer(option.IndexType, option.DirPath, option.SyncWrites),
		fileLock:   fileLock,
		ticker:     time.NewTicker(option.TimeSync),
	}

	// 加载 merge 数据目录
	if err := db.loadMergeFiles(); err != nil {
		return nil, err
	}

	// 加载数据文件
	fileIds, err := db.loadDataFiles()
	if err != nil {
		return nil, err
	}

	// B+ 树索引不需要从数据文件中加载
	if option.IndexType != index.BPTree {
		// 从 hint 索引文件中加载索引
		if err := db.loadIndexFromHintFile(); err != nil {
			return nil, err
		}

		// 加载数据内存索引
		if err := db.loadIndexFromDataFiles(fileIds); err != nil {
			return nil, err
		}

		// 加载完成后，重置 IO 类型为标准文件 IO
		if db.opt.MMapAtStartUp {
			if err := db.resetIOType(); err != nil {
				return nil, err
			}
		}
	}

	// B+ 树需要取出当前事务序列号，并且记录活跃文件偏移
	if option.IndexType == index.BPTree {
		if err := db.loadSeqNo(); err != nil {
			return nil, err
		}
		if db.activeFile != nil {
			size, err := db.activeFile.IOManager.Size()
			if err != nil {
				return nil, err
			}
			db.activeFile.WriteOff = size
		}
	}

	// 启动自动持久化数据文件协程
	go db.syncDataFileWithTime()

	return db, nil
}

// Stat 返回数据库的相关信息
func (d *DB) Stat() *Stat {
	d.mtx.RLock()
	defer d.mtx.RUnlock()

	var dataFileSize = uint(len(d.olderFiles))
	if d.activeFile != nil {
		dataFileSize += 1
	}

	dataDiskSize, err := utils.DirSize(d.opt.DirPath)
	if err != nil {
		panic(fmt.Sprintf("failed to get dir size: %v", err))
		return nil
	}

	// 获取数据目录所在磁盘的剩余可用空间大小
	availableDiskSize, _ := utils.AvailableDiskSize(d.opt.DirPath)

	return &Stat{
		KeyNum:            uint(d.index.Size()),
		DataFileNum:       dataFileSize,
		ReclaimableSize:   d.invalidDataSize,
		DataDiskSize:      dataDiskSize,
		AvailableDiskSize: int64(availableDiskSize),
	}
}

func (d *DB) Get(key []byte) ([]byte, error) {
	d.mtx.RLock()
	defer d.mtx.RUnlock()
	// 判断数据有效
	if len(key) == 0 {
		return nil, ErrKeyIsEmpty
	}
	// 获取数据索引信息
	logRecordPos := d.index.Get(key)
	// 获取实际value
	return d.getValueByPosition(logRecordPos)
}

// Put 写入 k/v 数据，key 不为空
func (d *DB) Put(key []byte, value []byte) error {
	// 判断 key 是否有效
	if len(key) == 0 {
		return ErrKeyIsEmpty
	}

	// 构建记录
	logRecord := &data.LogRecord{
		Key:   logRecordKeyWithSeq(key, nonTransactionSeqNo),
		Value: value,
		Type:  data.LogRecordNormal,
	}

	// 数据追加到当前活跃文件
	pos, err := d.appendLogRecordWithLock(logRecord)
	if err != nil {
		return err
	}

	// 更新索引
	oldPos := d.index.Put(key, pos)
	if oldPos != nil {
		d.invalidDataSize += int64(oldPos.Size)
	}
	return nil
}

func (d *DB) Delete(key []byte) error {
	// 判断key的有效性
	if len(key) == 0 {
		return ErrKeyIsEmpty
	}
	// 判断key是否存在
	if pos := d.index.Get(key); pos == nil {
		return nil
	}

	// 构造数据记录
	logRecord := &data.LogRecord{
		Key:  logRecordKeyWithSeq(key, nonTransactionSeqNo),
		Type: data.LogRecordDeleted,
	}

	// 插入数据文件
	pos, err := d.appendLogRecordWithLock(logRecord)
	if err != nil {
		return err
	}
	// 当前记录标记为可删除，即可被 merge 的无效数据
	d.invalidDataSize += int64(pos.Size)

	// 删除数据索引
	oldPos, ok := d.index.Delete(key)
	if !ok {
		return ErrIndexDeleteFailed
	}
	if oldPos != nil {
		d.invalidDataSize += int64(oldPos.Size)
	}
	return nil
}

// ListKeys 获取所有 keys
func (d *DB) ListKeys() [][]byte {
	it := d.index.Iterator(false)
	defer it.Close()
	keys := make([][]byte, d.index.Size())
	var idx = 0
	for it.Rewind(); it.Valid(); it.Next() {
		keys[idx] = it.Key()
		idx++
	}
	return keys
}

// Fold 遍历所有数据，执行用户自定义函数，遇到 false 终止执行
func (d *DB) Fold(fn func(k, v []byte) bool) error {
	d.mtx.RLock()
	defer d.mtx.RUnlock()

	it := d.index.Iterator(false)
	defer it.Close()
	for it.Rewind(); it.Valid(); it.Next() {
		value, err := d.getValueByPosition(it.Value())
		if err != nil {
			return err
		}

		if !fn(it.Key(), value) {
			break
		}
	}
	return nil
}

// Sync 持久化数据库
func (d *DB) Sync() error {
	if d.activeFile == nil {
		return nil
	}
	d.mtx.Lock()
	defer d.mtx.Unlock()

	if err := d.activeFile.Sync(); err != nil {
		return err
	}
	return nil
}

// Close 关闭数据库引擎
func (d *DB) Close() error {
	defer func() {
		// 关闭打点器
		d.ticker.Stop()
		// 释放文件锁
		if err := d.fileLock.Unlock(); err != nil {
			panic(err)
		}
	}()
	if d.activeFile == nil {
		// 尽管没使用数据库，但是启动的 B+ 树引擎实例需要关闭
		if err := d.index.Close(); err != nil {
			return err
		}
		return nil
	}
	d.mtx.Lock()
	defer d.mtx.Unlock()

	// 关闭索引，主要 B+ 树的 DB 实例需要关闭
	if err := d.index.Close(); err != nil {
		return err
	}

	// 保存事务序列号文件
	if err := d.saveSeqNoFile(); err != nil {
		return err
	}

	// 关闭活跃数据文件
	if err := d.activeFile.Close(); err != nil {
		return err
	}

	// 关闭旧数据文件
	for _, file := range d.olderFiles {
		if err := file.Close(); err != nil {
			return err
		}
	}
	return nil
}

func (d *DB) BackUp(dir string) error {
	d.mtx.Lock()
	defer d.mtx.Unlock()
	return utils.CopyDir(d.opt.DirPath, dir, []string{fileLockName})
}

// getValueByPosition 根据数据索引信息，从文件中读取实际存储 value 值
func (d *DB) getValueByPosition(logRecordPos *data.LogRecordPos) ([]byte, error) {
	if logRecordPos == nil {
		return nil, ErrKeyNotFound
	}

	// 获取数据记录文件
	var dataFile *data.DataFile
	if d.activeFile.FileId == logRecordPos.Fid {
		dataFile = d.activeFile
	} else {
		dataFile = d.olderFiles[logRecordPos.Fid]
	}

	// 判断数据文件存在
	if dataFile == nil {
		return nil, ErrDataFileNotFound
	}

	// 获取数据并返回
	logRecord, _, err := dataFile.ReadLogRecord(logRecordPos.Offset)
	if err != nil {
		return nil, err
	}

	// 如果记录标记删除，返回key不存在
	if logRecord.Type == data.LogRecordDeleted {
		return nil, ErrKeyNotFound
	}
	return logRecord.Value, nil
}

// appendLogRecordWithLock 追加数据前加锁
func (d *DB) appendLogRecordWithLock(logRecord *data.LogRecord) (*data.LogRecordPos, error) {
	d.mtx.Lock()
	defer d.mtx.Unlock()
	return d.appendLogRecord(logRecord)
}

// appendLogRecord 追加数据到当前活跃文件
func (d *DB) appendLogRecord(logRecord *data.LogRecord) (*data.LogRecordPos, error) {
	// 判断当前的活跃文件，如果没有则创建
	if d.activeFile == nil {
		if err := d.setActiveFile(); err != nil {
			return nil, err
		}
	}

	// 写入数据编码
	encRecord, size := data.EncodeLogRecord(logRecord)
	// 判断数据写入是否超出活跃文件阈值
	if d.activeFile.WriteOff+size > d.opt.DataFileSize {
		// 持久化当前活跃文件数据
		if err := d.activeFile.Sync(); err != nil {
			return nil, err
		}

		// 当前文件转为旧文件
		d.olderFiles[d.activeFile.FileId] = d.activeFile

		// 打开新的数据文件
		if err := d.setActiveFile(); err != nil {
			return nil, err
		}
	}

	// 执行数据写入
	writeOff := d.activeFile.WriteOff
	if err := d.activeFile.Write(encRecord); err != nil {
		return nil, err
	}

	// 累积写入数据字节数
	d.bytesWrite += uint(size)

	// 根据配置自动持久化
	var needSync = d.opt.SyncWrites
	if !needSync && d.opt.BytesSync > 0 && d.bytesWrite >= d.opt.BytesSync {
		needSync = true
	}
	if needSync {
		if err := d.activeFile.Sync(); err != nil {
			return nil, err
		}
		// 清空累积值
		if d.bytesWrite > 0 {
			d.bytesWrite = 0
		}
	}

	pos := &data.LogRecordPos{
		Fid:    d.activeFile.FileId,
		Offset: writeOff,
		Size:   uint32(size),
	}
	return pos, nil
}

// setActiveFile 创建新的文件作为当前活跃文件
// 必须持有互斥锁
func (d *DB) setActiveFile() error {
	var initialFileId uint32 = 0
	if d.activeFile != nil {
		initialFileId = d.activeFile.FileId + 1
	}

	// 打开新的数据文件
	dataFile, err := data.OpenDataFile(d.opt.DirPath, initialFileId, fio.StandardFIO)
	if err != nil {
		return err
	}
	d.activeFile = dataFile
	return nil
}

// loadIndexFromDataFiles 加载数据文件记录置索引中
func (d *DB) loadIndexFromDataFiles(fileIds []uint32) error {
	// 判断是否有数据文件
	if len(fileIds) == 0 {
		return nil
	}

	// 查看是否发生过 merge
	hasMerge, nonMergeFileId := false, uint32(0)
	mergeFinFileName := filepath.Join(d.opt.DirPath, data.MergeFileFinished)
	// 如果 merge 完成记录文件存在
	if _, err := os.Stat(mergeFinFileName); err == nil {
		// 获取最近未参与 merge 数据文件 id
		nonMergeFileId, err = d.getNonMergeFileId(d.opt.DirPath)
		if err != nil {
			return err
		}
		hasMerge = true
	}

	// 内存索引更新
	updateIndex := func(key []byte, typ data.LogRecordType, pos *data.LogRecordPos) {
		var oldPos *data.LogRecordPos
		if typ == data.LogRecordDeleted {
			oldPos, _ = d.index.Delete(key)
			d.invalidDataSize += int64(pos.Size)
		} else {
			oldPos = d.index.Put(key, pos)
		}
		if oldPos != nil {
			d.invalidDataSize += int64(oldPos.Size)
		}
	}

	// 暂存事务数据
	transactionRecords := make(map[uint64][]*data.TransactionRecord)
	var currentSeqNo = nonTransactionSeqNo

	// 顺序遍历当前数据文件
	for _, fid := range fileIds {
		// merge 过的数据文件已经从 hint 索引文件中加载过数据索引信息了
		if hasMerge && fid < nonMergeFileId {
			continue
		}
		// 拿到当前数据文件
		var dataFile *data.DataFile
		if fid == d.activeFile.FileId {
			dataFile = d.activeFile
		} else {
			dataFile = d.olderFiles[fid]
		}

		// 循环读取数据记录
		var offset int64 = 0
		for {
			// 拿到数据记录
			logRecord, size, err := dataFile.ReadLogRecord(offset)
			if err != nil {
				if err == io.EOF {
					break
				}
				return err
			}

			// 创建数据索引
			logRecordPos := &data.LogRecordPos{
				Fid:    fid,
				Offset: offset,
				Size:   uint32(size),
			}

			// 解析 key，拿到事务序列号
			realKey, seqNo := parseLogRecordKey(logRecord.Key)
			if seqNo == nonTransactionSeqNo {
				// 不是 writebatch 提交，非事务操作，直接更新索引
				updateIndex(realKey, logRecord.Type, logRecordPos)
			} else {
				// 事务完成，对应 SeqNo 的数据可更新到内存索引中
				if logRecord.Type == data.LogRecordTxnFinished {
					for _, txnRecord := range transactionRecords[seqNo] {
						updateIndex(txnRecord.Record.Key, txnRecord.Record.Type, txnRecord.Pos)
					}
					delete(transactionRecords, seqNo)
				} else {
					// 事务未完成，但该数据是 WriteBatch 写入
					// 暂存于事务表中 transactionRecord
					logRecord.Key = realKey
					transactionRecords[seqNo] = append(transactionRecords[seqNo], &data.TransactionRecord{
						Record: logRecord,
						Pos:    logRecordPos,
					})
				}
			}

			if seqNo > currentSeqNo {
				currentSeqNo = seqNo
			}

			// 递增 offset，下一次从新的位置开始读取
			offset += size
		}

		if fid == d.activeFile.FileId {
			d.activeFile.WriteOff = offset
		}
	}

	// 更新最新事务序列号
	d.seqNo = currentSeqNo
	return nil
}

// loadDataFiles 加载数据文件，返回数据文件顺序fids
func (d *DB) loadDataFiles() ([]uint32, error) {
	// 读取目录下的所有文件
	dirEntries, err := os.ReadDir(d.opt.DirPath)
	if err != nil {
		return nil, err
	}

	// 提取文件前缀 fid
	fileIds := make([]uint32, 0)
	for _, entry := range dirEntries {
		if strings.HasSuffix(entry.Name(), data.DataFileSuffix) {
			names := strings.Split(entry.Name(), ".")
			fid, _ := strconv.ParseUint(names[0], 10, 32)
			fileIds = append(fileIds, uint32(fid))
		}
	}

	// 排序
	slices.Sort(fileIds)

	// 加载所有数据文件
	for i, fid := range fileIds {
		// 根据配置使用指定 IO 类型加载数据文件
		var ioType fio.FileIOType = fio.StandardFIO
		if d.opt.MMapAtStartUp {
			ioType = fio.MemoryMap
		}

		// 数据文件有可能损坏
		dataFile, err := data.OpenDataFile(d.opt.DirPath, fid, ioType)
		if err != nil {
			return fileIds, ErrFileDestroyCorrupted
		}
		// 最后一个作为活跃文件
		if i == len(fileIds)-1 {
			d.activeFile = dataFile
		} else {
			d.olderFiles[fid] = dataFile
		}
	}

	return fileIds, nil
}

// loadSeqNo 加载事务序列号文件
func (d *DB) loadSeqNo() error {
	fileName := filepath.Join(d.opt.DirPath, data.SeqNoFileName)
	if _, err := os.Stat(fileName); os.IsNotExist(err) {
		return nil
	}

	seqNoFile, err := data.OpenSeqNoFile(d.opt.DirPath)
	defer seqNoFile.Close()
	if err != nil {
		return err
	}
	record, _, err := seqNoFile.ReadLogRecord(0)
	if err != nil {
		return err
	}
	seqNo, err := strconv.ParseUint(string(record.Value), 10, 64)
	if err != nil {
		return err
	}
	d.seqNo = seqNo
	return nil
}

// saveSeqNoFile 保存事务序列号文件
func (d *DB) saveSeqNoFile() error {
	seqNoFile, err := data.OpenSeqNoFile(d.opt.DirPath)
	defer seqNoFile.Close()
	if err != nil {
		return err
	}
	record := &data.LogRecord{
		Key:   []byte(seqNoKey),
		Value: []byte(strconv.FormatUint(d.seqNo, 10)),
	}
	encRecord, _ := data.EncodeLogRecord(record)
	if err := seqNoFile.Write(encRecord); err != nil {
		return err
	}
	return nil
}

// syncDataFileWithTime 需要另开协程，根据配置时间自动持久化数据文件
func (d *DB) syncDataFileWithTime() {
	for {
		select {
		case <-d.ticker.C:
			if err := d.activeFile.Sync(); err != nil {
				panic(err)
			}
		}
	}
}

// resetIOType 重置 IO 类型为标准文件 IO
func (d *DB) resetIOType() error {
	if d.activeFile == nil {
		return nil
	}
	if err := d.activeFile.SetIOManager(d.opt.DirPath, fio.StandardFIO); err != nil {
		return err
	}
	for _, file := range d.olderFiles {
		if err := file.SetIOManager(d.opt.DirPath, fio.StandardFIO); err != nil {
			return err
		}
	}
	return nil
}

// checkOptions 检查配置文件
func checkOptions(option *Options) error {
	if option == nil {
		return ErrOptionIsEmpty
	}

	// 默认 64K 文件大小
	if option.DataFileSize <= 0 {
		option.DataFileSize = DataFile64K
	}

	// 默认每隔 1h 自动持久化
	if option.TimeSync <= 0 {
		option.TimeSync = time.Hour
	}

	// 默认无效数据至少超过 10% 才可以进行执行 merge
	if option.DataFileMergeRatio >= 1 {
		option.DataFileMergeRatio = 0.1
	}

	// 目录路径标准化
	option.DirPath = utils.NormalizationFilePathString(option.DirPath)
	//校验目录，不存在创建
	if _, err := os.Stat(option.DirPath); os.IsNotExist(err) {
		if err := os.MkdirAll(option.DirPath, os.ModePerm); err != nil {
			return err
		}
	}

	return nil
}
